package com.atguigu.gmall.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.app.func.AsyncJoinFunction;
import com.atguigu.gmall.bean.TradeProvinceOrderWindow;
import com.atguigu.gmall.util.ClickHouseUtil;
import com.atguigu.gmall.util.DateFormatUtil;
import com.atguigu.gmall.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author yhm
 * @create 2022-09-21 14:15
 */
public class DwsTradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取topic_db数据
        String topicName = "topic_db";
        String groupId = "dws_trade_province_order_window";
        DataStreamSource<String> topicDbStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));

        // TODO 4 过滤下单数据转换结构
        SingleOutputStreamOperator<TradeProvinceOrderWindow> jsonObjStream = topicDbStream.flatMap(new FlatMapFunction<String, TradeProvinceOrderWindow>() {
            @Override
            public void flatMap(String value, Collector<TradeProvinceOrderWindow> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String table = jsonObject.getString("table");
                String type = jsonObject.getString("type");
                JSONObject data = jsonObject.getJSONObject("data");
                if ("insert".equals(type) && "order_info".equals(table)) {
                    out.collect(TradeProvinceOrderWindow.builder()
                            .orderCount(1L)
                            .orderAmount(data.getDouble("total_amount"))
                            .provinceId(data.getString("province_id"))
                            .ts(jsonObject.getLong("ts") * 1000L)
                            .build());
                }
            }
        });

        // TODO 5 按照省份ID开窗聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> reduceBeanStream = jsonObjStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderWindow>() {
            @Override
            public long extractTimestamp(TradeProvinceOrderWindow element, long recordTimestamp) {
                return element.getTs();
            }
        })).keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1, TradeProvinceOrderWindow value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());

                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                }, new WindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeProvinceOrderWindow> input, Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow next = input.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });


        // TODO 6 维度关联省份名称
        SingleOutputStreamOperator<TradeProvinceOrderWindow> provinceBeanStream = AsyncDataStream.unorderedWait(reduceBeanStream, new AsyncJoinFunction<TradeProvinceOrderWindow>("dim_base_province".toUpperCase()) {
            @Override
            public String getKey(TradeProvinceOrderWindow input) {
                return input.getProvinceId();
            }

            @Override
            public void join(TradeProvinceOrderWindow input, JSONObject dimSkuInfo) {
                input.setProvinceName(dimSkuInfo.getString("name"));
            }
        }, 60L, TimeUnit.SECONDS);

        provinceBeanStream.print("province>>>");

        // TODO 7 写出到clickHouse中
        provinceBeanStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_trade_province_order_window values(?,?,?,?,?,?,?)"));

        // TODO 8 执行任务
        env.execute(groupId);

    }
}
